package extension;

import lib.ApiClient ;
import common.Utils;
import com.github.luben.zstd.Zstd;
import org.apache.flink.avro.shaded.org.apache.avro.Schema;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericData;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericDatumReader;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericRecord;
import org.apache.flink.avro.shaded.org.apache.avro.io.BinaryDecoder;
import org.apache.flink.avro.shaded.org.apache.avro.io.DatumReader;
import org.apache.flink.avro.shaded.org.apache.avro.io.DecoderFactory;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.conversion.RowRowConverter;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Base64;
import java.util.Date;
import java.util.TimeZone;

@FunctionHint(output = @DataTypeHint("ROW<" +
                    "platform STRING\n, dt STRING," +
                    "header ROW<" +
                        "basic ROW<" +
                            "message_id STRING, project_id STRING, app_id STRING, tenant_id STRING, upload_time STRING," +
                            "apex_id STRING, session_id STRING, app_version STRING, app_package_id STRING, carrier STRING, mini_program_scene STRING," +
                            "market_name STRING" +
                        ">," +
                        "sdk_info ROW<sdk_version STRING, sdk_type STRING>," +
                        "device_info ROW<" +
                            "device_id STRING, device_model STRING, device_brand STRING, device_type STRING, device_uuid STRING," +
                            "android_id STRING, imei STRING, mac STRING, user_agent STRING, idfa STRING," +
                            "idfv STRING, oaid STRING" +
                        ">," +
                        "screen ROW<screen_height STRING, screen_width STRING>," +
                        "os ROW<os_system STRING, os_version STRING>," +
                        "browser ROW<browser_type STRING, browser_version STRING>," +
                        "wechat ROW<wechat_open_id STRING, wechat_union_id STRING>," +
                        "referrer_info ROW<referrer_url STRING, referrer_host STRING>," +
                        "utm_info ROW<" +
                            "utm_matching_type STRING, utm_source STRING, utm_medium STRING, utm_term STRING, utm_content STRING," +
                            "utm_campaign STRING" +
                        ">" +
                    ">," +
                    "basic ROW<" +
                        "event_code STRING, event_name STRING, event_type STRING, send_type STRING, page_type STRING," +
                        "ts STRING, network_type STRING, time_since_last_operation STRING, event_code_md5 STRING" +
                    ">," +
                    "business_info ROW<" +
                        "user_id STRING," +
                        "extensions MAP<STRING, STRING>" +
                    ">," +
                    "location_info ROW<" +
                        "ip STRING," +
                        "latitude STRING," +
                        "longitude STRING," +
                        "province STRING," +
                        "city STRING," +
                        "country STRING," +
                        "province_id STRING," +
                        "city_id STRING," +
                        "county STRING" +
                    ">," +
                    "page_info ROW<" +
                        "current_page_id STRING," +
                        "current_page_name STRING," +
                        "current_page_url STRING," +
                        "current_page_host STRING," +
                        "current_page_query STRING," +
                        "current_page_path STRING," +
                        "source_page_id STRING," +
                        "source_page_name STRING," +
                        "source_page_url STRING," +
                        "source_page_host STRING," +
                        "source_page_query STRING," +
                        "source_page_path STRING" +
                    ">," +
                    "size_info ROW<" +
                        "visible_window_width STRING," +
                        "visible_window_height STRING," +
                        "page_height STRING," +
                        "page_width STRING" +
                    ">" +
    ">"))
public class ParseDcEventCustom extends TableFunction<Row> {
    boolean inited = false;
    Schema batch_schema = null;
    DatumReader<Object> batch_reader = null;

    public static String parseChinaDt(String s) {
        long tsL = Long.parseLong(s);
        Date dt = new Date(tsL);
        SimpleDateFormat bjSdf = new SimpleDateFormat("yyyyMMdd");
        bjSdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        return bjSdf.format(dt);
    }

    public static Object avroGet(GenericRecord avro, String field) {
        try {
            return avro.get(field);
        } catch (Exception e) {
            return null;
        }
    }

    public static DataType getDataType() {
        DataType headerDataType = DataTypes.ROW(
            DataTypes.FIELD("basic", DataTypes.ROW(
                DataTypes.FIELD("message_id", DataTypes.STRING())
                , DataTypes.FIELD("project_id", DataTypes.STRING())
                , DataTypes.FIELD("app_id", DataTypes.STRING())
                , DataTypes.FIELD("tenant_id", DataTypes.STRING())
                , DataTypes.FIELD("upload_time", DataTypes.STRING())

                , DataTypes.FIELD("apex_id", DataTypes.STRING())
                , DataTypes.FIELD("session_id", DataTypes.STRING())
                , DataTypes.FIELD("app_version", DataTypes.STRING())
                , DataTypes.FIELD("app_package_id", DataTypes.STRING())
                , DataTypes.FIELD("carrier", DataTypes.STRING())
                , DataTypes.FIELD("mini_program_scene", DataTypes.STRING())

                , DataTypes.FIELD("market_name", DataTypes.STRING())
            ))
            , DataTypes.FIELD("sdk_info", DataTypes.ROW(
                DataTypes.FIELD("sdk_version", DataTypes.STRING())
                , DataTypes.FIELD("sdk_type", DataTypes.STRING())
            ))
            , DataTypes.FIELD("device_info", DataTypes.ROW(
                DataTypes.FIELD("device_id", DataTypes.STRING())
                , DataTypes.FIELD("device_model", DataTypes.STRING())
                , DataTypes.FIELD("device_brand", DataTypes.STRING())
                , DataTypes.FIELD("device_type", DataTypes.STRING())
                , DataTypes.FIELD("device_uuid", DataTypes.STRING())

                , DataTypes.FIELD("android_id", DataTypes.STRING())
                , DataTypes.FIELD("imei", DataTypes.STRING())
                , DataTypes.FIELD("mac", DataTypes.STRING())
                , DataTypes.FIELD("user_agent", DataTypes.STRING())
                , DataTypes.FIELD("idfa", DataTypes.STRING())

                , DataTypes.FIELD("idfv", DataTypes.STRING())
                , DataTypes.FIELD("oaid", DataTypes.STRING())
            ))
            , DataTypes.FIELD("screen", DataTypes.ROW(
                DataTypes.FIELD("screen_height", DataTypes.STRING())
                , DataTypes.FIELD("screen_width", DataTypes.STRING())
            ))
            , DataTypes.FIELD("os", DataTypes.ROW(
                DataTypes.FIELD("os_system", DataTypes.STRING())
                , DataTypes.FIELD("os_version", DataTypes.STRING())
            ))
            , DataTypes.FIELD("browser", DataTypes.ROW(
                DataTypes.FIELD("browser_type", DataTypes.STRING())
                , DataTypes.FIELD("browser_version", DataTypes.STRING())
            ))
            , DataTypes.FIELD("wechat", DataTypes.ROW(
                DataTypes.FIELD("wechat_open_id", DataTypes.STRING())
                , DataTypes.FIELD("wechat_union_id", DataTypes.STRING())
            ))
            , DataTypes.FIELD("referrer_info", DataTypes.ROW(
                DataTypes.FIELD("referrer_url", DataTypes.STRING())
                , DataTypes.FIELD("referrer_host", DataTypes.STRING())
            ))
            , DataTypes.FIELD("utm_info", DataTypes.ROW(
                DataTypes.FIELD("utm_matching_type", DataTypes.STRING())
                , DataTypes.FIELD("utm_source", DataTypes.STRING())
                , DataTypes.FIELD("utm_medium", DataTypes.STRING())
                , DataTypes.FIELD("utm_term", DataTypes.STRING())
                , DataTypes.FIELD("utm_content", DataTypes.STRING())
                , DataTypes.FIELD("utm_campaign", DataTypes.STRING())
            ))
        );
        DataType dataType = DataTypes.ROW(
            DataTypes.FIELD("platform", DataTypes.STRING())
            , DataTypes.FIELD("dt", DataTypes.STRING())
            , DataTypes.FIELD("header", headerDataType)
            , DataTypes.FIELD("basic", DataTypes.ROW(
                DataTypes.FIELD("event_code", DataTypes.STRING())
                , DataTypes.FIELD("event_name", DataTypes.STRING())
                , DataTypes.FIELD("event_type", DataTypes.STRING())
                , DataTypes.FIELD("send_type", DataTypes.STRING())
                , DataTypes.FIELD("page_type", DataTypes.STRING())

                , DataTypes.FIELD("ts", DataTypes.STRING())
                , DataTypes.FIELD("network_type", DataTypes.STRING())
                , DataTypes.FIELD("time_since_last_operation", DataTypes.STRING())
                , DataTypes.FIELD("event_code_md5", DataTypes.STRING())
            ))
            , DataTypes.FIELD("business_info", DataTypes.ROW(
                DataTypes.FIELD("user_id", DataTypes.STRING()),
                DataTypes.FIELD("extensions", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
            ))
            , DataTypes.FIELD("location_info", DataTypes.ROW(
                DataTypes.FIELD("ip", DataTypes.STRING())
                , DataTypes.FIELD("latitude", DataTypes.STRING())
                , DataTypes.FIELD("longitude", DataTypes.STRING())
                , DataTypes.FIELD("province", DataTypes.STRING())
                , DataTypes.FIELD("city", DataTypes.STRING())

                , DataTypes.FIELD("country", DataTypes.STRING())
                , DataTypes.FIELD("province_id", DataTypes.STRING())
                , DataTypes.FIELD("city_id", DataTypes.STRING())
                , DataTypes.FIELD("county", DataTypes.STRING())
            ))
            , DataTypes.FIELD("page_info", DataTypes.ROW(
                DataTypes.FIELD("current_page_id", DataTypes.STRING())
                , DataTypes.FIELD("current_page_name", DataTypes.STRING())
                , DataTypes.FIELD("current_page_url", DataTypes.STRING())
                , DataTypes.FIELD("current_page_host", DataTypes.STRING())
                , DataTypes.FIELD("current_page_query", DataTypes.STRING())
                , DataTypes.FIELD("current_page_path", DataTypes.STRING())
                , DataTypes.FIELD("source_page_id", DataTypes.STRING())
                , DataTypes.FIELD("source_page_name", DataTypes.STRING())
                , DataTypes.FIELD("source_page_url", DataTypes.STRING())
                , DataTypes.FIELD("source_page_host", DataTypes.STRING())
                , DataTypes.FIELD("source_page_query", DataTypes.STRING())
                , DataTypes.FIELD("source_page_path", DataTypes.STRING())
            ))
            , DataTypes.FIELD("size_info", DataTypes.ROW(
                DataTypes.FIELD("visible_window_width", DataTypes.STRING())
                , DataTypes.FIELD("visible_window_height", DataTypes.STRING())
                , DataTypes.FIELD("page_height", DataTypes.STRING())
                , DataTypes.FIELD("page_width", DataTypes.STRING())
            ))
        );

        return dataType;
    }

    public static RowData parseHeader(GenericRecord avroHeader) {
        String headerBasicFields = "message_id,project_id,app_id,tenant_id,upload_time,apex_id,session_id,app_version, app_package_id,carrier,mini_program_scene,market_name";
        String headerSdkInfoFields = "sdk_version,sdk_type";
        String headerDeviceInfoFields = "device_id,device_model,device_brand,device_type,device_uuid,android_id,imei,mac,user_agent,idfa,idfv,oaid";
        String headerScreenFields = "screen_height,screen_width";
        String headerOsFields = "os_system,os_version";
        String headerBrowserFields = "browser_type,browser_version";
        String headerWechatFields = "wechat_open_id,wechat_union_id";
        String headerReferrerInfoFields = "referrer_url,referrer_host";
        String headerUtmInfoFields = "utm_matching_type,utm_source,utm_medium,utm_term,utm_content,utm_campaign";

        // System.out.println("avroHeader:" + avroHeader) ;
        RowData header = Utils.mkRow(avroHeader, "",
            Utils.mkRow(avroHeader.get("basic"), headerBasicFields),
            Utils.mkRow(avroHeader.get("sdk_info"), headerSdkInfoFields),
            Utils.mkRow(avroHeader.get("device_info"), headerDeviceInfoFields),
            Utils.mkRow(avroHeader.get("screen"), headerScreenFields),
            Utils.mkRow(avroHeader.get("os"), headerOsFields),
            Utils.mkRow(avroHeader.get("browser"), headerBrowserFields),
            // Utils.mkRow(avroHeader.get("wechat"), headerWechatFields),
            Utils.mkRow(avroGet(avroHeader, "wechat"), headerWechatFields),
            Utils.mkRow(avroHeader.get("referrer_info"), headerReferrerInfoFields),
            Utils.mkRow(avroHeader.get("utm_info"), headerUtmInfoFields));
        return header;

    }

    public static RowData parseEvent(GenericRecord sdkEvent, String platformName, String dt, RowData header) {
        if (!sdkEvent.hasField("is_custom")) {
            return null;
        }
        String basicInfoFields = "event_code,event_name,event_type,send_type,page_type,ts,network_type,time_since_last_operation,event_code_md5,browse_time";
        String coordinatorInfoFields = "page_x,page_y,client_x,client_y";
        String elementInfoFields = "element_id,element_name,element_target_url,element_type,element_content,element_selector,element_position,element_index,element_tag_name,element_class,element_path";
        String businessInfoFields = "user_id,extensions";
        String locationInfoFields = "ip,latitude,longitude,province,city,country,province_id,city_id,county";
        String pageInfoFields = "current_page_id,current_page_name,current_page_url,current_page_host,current_page_query,current_page_path,source_page_id,source_page_name,source_page_url,source_page_host,source_page_query,source_page_path";
        String sizeInfoFields = "visible_window_width,visible_window_height,page_height,page_width";
        return GenericRowData.of(
            StringData.fromString(platformName), StringData.fromString(dt),
            header,
            Utils.mkRow(sdkEvent.get("basic"), basicInfoFields),
            Utils.mkRow(sdkEvent.get("business_info"), businessInfoFields),
            Utils.mkRow(sdkEvent.get("location_info"), locationInfoFields),
            Utils.mkRow(sdkEvent.get("page_info"), pageInfoFields),
            Utils.mkRow(avroGet(sdkEvent, "size_info"), sizeInfoFields)
        );
    }

    public void init(String metaServerNodes) throws Exception {
        if (inited) {
            return;
        }
        String schema_s = ApiClient.getMetaSchema(metaServerNodes) ;
        batch_schema = new Schema.Parser().parse(schema_s);
        batch_reader = new GenericDatumReader<Object>(batch_schema);
        inited = true;
    }

    public static String extractPlatformName(String schemaName) {
        String result = "unknown" ;
        if (schemaName.equals("DcSdkAndroid")) {
            result = "android" ;
        } else if (schemaName.equals("DcSdkIos")) {
            result = "ios" ;
        } else if (schemaName.equals("DcSdkWeb")) {
            result = "web" ;
        } else if (schemaName.equals("DcSdkWechat")) {
            result = "wechat" ;
        } else if (schemaName.equals("DcSdkH5")) {
            result = "h5" ;
        }
        return result ;
    }

    public void eval(String metaServerNodes, String b64) throws Exception {
        try {
            init(metaServerNodes);
            byte[] avro_zstd = Base64.getDecoder().decode(b64.getBytes());
            byte[] avro = Zstd.decompress(avro_zstd, avro_zstd.length * 50);

            BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(avro, null);
            GenericData.Array<GenericRecord> bulk = (GenericData.Array<GenericRecord>) batch_reader.read(null, binaryDecoder);

            for (GenericRecord sdkRecord : bulk) {
                String platformName = extractPlatformName(sdkRecord.getSchema().getName());
                GenericRecord avroHeader = (GenericRecord) sdkRecord.get("common");

                RowData header = parseHeader(avroHeader);
                for (GenericRecord sdkEvent : (GenericData.Array<GenericRecord>) sdkRecord.get("events")) {
                    GenericRecord basicInfo = (GenericRecord)sdkEvent.get("basic") ;
                    String dt = parseChinaDt(basicInfo.get("ts").toString());

                    RowData headerEvent = parseEvent(sdkEvent, platformName, dt, header);
                    if (headerEvent == null)
                        continue;
                    Row headerEventRow = RowRowConverter.create(getDataType()).toExternal(headerEvent);
                    collect(headerEventRow);
                }
            }

        } catch (Exception e) {
            System.out.println("eror data:" + b64) ;
            e.printStackTrace();
            throw e;
        }
    }

}

